package com.spx.chapter08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * create by undeRdoG on  2022-05-04  14:48
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class BillCheckExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 来自app的支付日志
        SingleOutputStreamOperator<Tuple3<String, String, Long>> appStream = env.fromElements(
                Tuple3.of("order-1", "app", 1000L),
                Tuple3.of("order-2", "app", 2000L),
                Tuple3.of("order-3", "app", 2000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Tuple3<String, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                            @Override
                            public long extractTimestamp(Tuple3<String, String, Long> element, long recordTimestamp) {
                                return element.f2;

                            }
                        })
        );

        // 来自第三方支付平台日志

        SingleOutputStreamOperator<Tuple4<String, String, String, Long>> thirdPartyStream = env.fromElements(
                Tuple4.of("order-1", "third-party", "success", 3000L),
                Tuple4.of("order-3", "third-party", "success", 4000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Tuple4<String, String, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple4<String, String, String, Long>>() {
                            @Override
                            public long extractTimestamp(Tuple4<String, String, String, Long> element, long recordTimestamp) {
                                return element.f3;
                            }
                        })
        );

        //检测同一支付单是否在两条流中匹配

        appStream.connect(thirdPartyStream)
                .keyBy(data -> data.f0, data -> data.f0)
                // 需要在一定时间范围内检测，需要定时器，直接使用ProcessFunction
                .process(new OrderMatchResult())
                .print();


        env.execute();

    }


    public static class OrderMatchResult extends CoProcessFunction<Tuple3<String, String, Long>, Tuple4<String, String, String, Long>, String> {

        // 之前对两条流已经进行了keyBy，那么进入到同一个处理实例的数据一定是相同的key

        // 定义状态变量，用来保存已经到达的数据
        private ValueState<Tuple3<String, String, Long>> appEvent;

        private ValueState<Tuple4<String, String, String, Long>> thirdPartyEvent;

        @Override
        public void open(Configuration parameters) throws Exception {
            this.appEvent = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple3<String, String, Long>>("appState", Types.TUPLE(Types.STRING, Types.STRING, Types.LONG))
            );

            this.thirdPartyEvent = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple4<String, String, String, Long>>("thirdPartyEvent", Types.TUPLE(Types.STRING, Types.STRING, Types.STRING, Types.LONG))
            );
        }

        @Override
        public void processElement1(Tuple3<String, String, Long> value, Context ctx, Collector<String> out) throws Exception {
            // appEvent数据到达，查看另一条数据流的状态
            if (thirdPartyEvent.value() != null){
                out.collect("对账成功" + value + " " + thirdPartyEvent.value());
                // 清空状态
                thirdPartyEvent.clear();
            }else {
                //更新自己的状态
                appEvent.update(value);
                // 注册一个5S的定时器，开始等待另一条流的事件
                ctx.timerService().registerEventTimeTimer(value.f2 + 5000L);
            }
        }

        @Override
        public void processElement2(Tuple4<String, String, String, Long> value, Context ctx, Collector<String> out) throws Exception {
            if (appEvent.value() != null){
                out.collect("对账成功" + value + " " + appEvent.value());
                appEvent.clear();
            }else {
                thirdPartyEvent.update(value);

                ctx.timerService().registerEventTimeTimer(value.f3 + 5000L);
            }

        }


        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            if (appEvent.value() != null){
                out.collect("对账失败" + appEvent.value() + " " + "第三方数据未到");
            }

            if (thirdPartyEvent != null){
                out.collect("对账失败" + thirdPartyEvent.value() + " " + "App数据未到");
            }

            appEvent.clear();

            thirdPartyEvent.clear();
        }
    }





}
